// rpc.cc
// Copyright (c) 1992-1993 The Regents of the University of California.
// All rights reserved.  See copyright.h for copyright notice and limitation
// of liability and disclaimer of warranty provisions.

#include "copyright.h"

#include "system.h"
#include "network.h"
#include "post.h"
#include "interrupt.h"

extern "C" { int bzero(char *, int); };

RPCLock rpclock[Max_RPC_LOCKNUM];
Table  *    rpclockTable=new Table(Max_RPC_LOCKNUM);
Lock * rpclockLock=new Lock("rpclockLock");

RPCCV rpcCV[Max_RPC_CVNUM];
Table   *   rpcCVTable=new Table(Max_RPC_CVNUM);
Lock * rpcCVLock=new Lock("rpcCVLock");

RPCMV rpcMV[Max_RPC_MVNUM];
Table   *   rpcMVTable=new Table(Max_RPC_MVNUM);

//global varables for communication
//used for the  client
ServerInfo * myserver=new ServerInfo;
ClientInfo  * clientinfo= new ClientInfo;

//used for the server
ServerInfo * serverinfo= new ServerInfo;
ClientInfo authclients[MAX_USERNUM];
int curAutUsrNum=0;

RpcReqMessage reqArray[MAX_REPLAYLISTNUM];
Table  *  reqTable=new Table(MAX_REPLAYLISTNUM);

RpcRplyMessage replayArray[MAX_REPLAYLISTNUM];
Table  *  rplyTable=new Table(MAX_REPLAYLISTNUM);

RpcPendingRequest pendingRequest[MAX_PENDING_REQUEST];
Table  *  prTable=new Table(MAX_PENDING_REQUEST);

RpcDYHMessage dyhArray[MAX_REPLAYLISTNUM];
Table  *  dyhTable=new Table(MAX_REPLAYLISTNUM);

//int serverNumber=1;//TODO:move to main.cc

Timer * pingTimer;
int getLockByName(char * name){
    int i=0;
    for(i=0;i<Max_RPC_LOCKNUM;i++){
        if(0==strcmp(name,rpclock[i].name))
             break;
    }
    if(Max_RPC_LOCKNUM==i)
        return -1;
    else {
        DEBUGINFO('r', "get lock %d, %s \n", i,rpclock[i].name)
        return i;
    }
}

int getCVByName(char * name){
    int i=0;
    for(i=0;i<Max_RPC_CVNUM;i++){
        if(0==strcmp(name,rpcCV[i].name))
             break;
    }
    if(Max_RPC_CVNUM==i)
        return -1;
    else 
        return i;
}

int getMVByName(char * name){
    int i=0;
    for(i=0;i<Max_RPC_MVNUM;i++){
       // printf("name: %s, rpcMV[%d].name:%s \n",name,i,rpcMV[i].name);
        if(0==strcmp(name,rpcMV[i].name))
             break;
    }
    if(Max_RPC_MVNUM==i)
        return -1;
    else 
        return i;
}


//the sever wait for the request of the client
void waitForRequst(RpcReqMessage * request)
{
    DEBUGINFO('r',"before wait , last received type %d, intparm1 %d, inytparm2 %d, request string %s, from m %d, box %d, ping %d\n",request->reqType,request->reqIntParam1,request->reqIntParam2,request->reqStringParam,request->requestFromAddr,request->requestFromMail, request->pingId);
    PacketHeader inPktHdr;
    MailHeader inMailHdr;
    char databuffer[MaxMailSize];
    postOffice->Receive(currentThread->mailId, &inPktHdr, &inMailHdr, databuffer);
    bcopy(databuffer,( char *)request, sizeof(RpcReqMessage));
    //request->requestFromAddr=inPktHdr.from;
    //request->requestFromMail=inMailHdr.from;
    DEBUGINFO('r',"received type %d, intparm1 %d, inytparm2 %d, request string %s, from m %d, box %d, ping %d\n",request->reqType,request->reqIntParam1,request->reqIntParam2,request->reqStringParam,request->requestFromAddr,request->requestFromMail, request->pingId);
    
}

//the sever wait for the request of the client
void waitForDYH(RpcDYHMessage *dyh)
{
    PacketHeader inPktHdr;
    MailHeader inMailHdr;
    char databuffer[MaxMailSize];
    postOffice->Receive(currentThread->mailId, &inPktHdr, &inMailHdr, databuffer);
    bcopy(databuffer,( char *)dyh, sizeof(RpcDYHMessage));
    dyh->FromAddr=inPktHdr.from;
    dyh->FromMail=inMailHdr.from;
    DEBUGINFO('r',"received dyh type from m %d, box %d, dyh->type %d, dyh->name %s , dyh->msgtype %d, dyh->reply %d\n",dyh->FromAddr,dyh->FromMail, dyh->type, dyh->name, dyh->msgtype, dyh->reply);
}
//the server reply to the client
bool replyRequest(RpcRplyMessage* rplymsg)
{
    PacketHeader outPktHdr;
    MailHeader outMailHdr;
    char databuffer[MaxMailSize];
    // construct packet, mail header for original message
    // To: destination machine, mailbox 0
    // From: our machine, reply to: mailbox 1
    outPktHdr.from=currentThread->haddr;
    outPktHdr.to = rplymsg->replyToAddr;
    outMailHdr.to = rplymsg->replyToMail;
    outMailHdr.from = 0;
    outMailHdr.length=sizeof(RpcRplyMessage);
    bcopy( (char *) rplymsg, databuffer, sizeof(RpcRplyMessage));
    bool success = postOffice->Send(outPktHdr, outMailHdr,databuffer);
    DEBUGINFO('r',"reply request type %d, intparm %d to %d, box %d\n",rplymsg->ansType,rplymsg->ansIntParam1,outPktHdr.to,outMailHdr.to);
    return success;

}
//send the request to the registered server
bool  sendRequest(RpcReqMessage * rrmsg, int serverHaddr, int serverMail)
{
    PacketHeader outPktHdr;
    MailHeader outMailHdr;
    char databuffer[MaxMailSize];
    // step 1; init the packat
    outPktHdr.to = serverHaddr;
    outMailHdr.to = serverMail;
    outPktHdr.from = netname;
    outMailHdr.from = currentThread->mailId;
    outMailHdr.length=sizeof(RpcReqMessage);
    //step 2: set the from field of the request msg
    rrmsg->requestFromAddr=netname;
    rrmsg->requestFromMail=outMailHdr.from;
     //step 3: send the request to the server
     bcopy((char *)rrmsg,databuffer,sizeof(RpcReqMessage));
    bool success = postOffice->Send(outPktHdr, outMailHdr,databuffer);
    DEBUGINFO('r',"send request type %d, intparm1 %d, inytparm2 %d, request string %s, from pindId %d, to m %d, box %d %s\n",rrmsg->reqType,rrmsg->reqIntParam1,rrmsg->reqIntParam2,rrmsg->reqStringParam, rrmsg->pingId, outPktHdr.to,outMailHdr.to,rrmsg->reqStringParam);
    return success;
}

//send the request to the registered server
bool  forwardRequest(RpcReqMessage * rrmsg, int serverHaddr, int serverMail)
{
    PacketHeader outPktHdr;
    MailHeader outMailHdr;
    char databuffer[MaxMailSize];
    // step 1; init the packat
    outPktHdr.to = serverHaddr;
    outMailHdr.to = serverMail;
    outPktHdr.from = currentThread->haddr;
    outMailHdr.from = currentThread->mailId;
    outMailHdr.length=sizeof(RpcReqMessage);
    //step 2: set the from field of the request msg
     //step 3: send the request to the server
     bcopy((char *)rrmsg,databuffer,sizeof(RpcReqMessage));
    bool success = postOffice->Send(outPktHdr, outMailHdr,databuffer);

    DEBUGINFO('r',"forward sent type %d, intparm1 %d, inytparm2 %d, request string %s, from pindId %d, to m %d, box %d %s\n",rrmsg->reqType,rrmsg->reqIntParam1,rrmsg
->reqIntParam2,rrmsg->reqStringParam, rrmsg->pingId, outPktHdr.to,outMailHdr.to,rrmsg->reqStringParam);
    return success;
}

//get the reply from the server
void getReply(RpcRplyMessage * reply)
{
    PacketHeader inPktHdr;
    MailHeader inMailHdr;
    char buffer[MaxMailSize];
    postOffice->Receive(currentThread->mailId, &inPktHdr, &inMailHdr, buffer);
    bcopy(buffer,( char *)reply, sizeof(RpcRplyMessage));
    DEBUGINFO('r',"reply type %d, intparm1 %d,isvalid %d from m %d, box %d; to m %d box %d",reply->ansType,reply->ansIntParam1,reply->isValid, inPktHdr.from, inMailHdr.from, reply->replyToAddr, reply->replyToMail);
}

void setReply(RpcRplyMessage * src,RpcRplyMessage * dst)
{
    src->ansIntParam1=dst->ansIntParam1;
    src->ansType=dst->ansType;
    src->ansVersion=dst->ansVersion;
    src->isValid=dst->isValid;
    src->replyToAddr=dst->replyToAddr;
    src->replyToMail=dst->replyToMail;
    src->pingId=dst->pingId;
}

void setReq(RpcReqMessage * src,RpcReqMessage * dst)
{
    src->reqType=dst->reqType;
    src->requestFromAddr=dst->requestFromAddr;
    src->requestFromMail=dst->requestFromMail;
}

void setDYH(RpcDYHMessage * src, RpcDYHMessage * dst) {
    strcpy(src->name, dst->name);
    src->msgtype = dst->msgtype;
    src->reply = dst->reply;
    src->timestamp = dst->timestamp;
    src->FromAddr = dst->FromAddr;
    src->FromMail = dst->FromMail;
    src->type = dst->type;
    src->lockId = dst->lockId;
    src->isCVForLock = dst->isCVForLock;
}

bool isOwner(RPCLock * rplock, RpcReqMessage * request)
{
    DEBUGINFO('r', "check is owner rplockowner: hdr %d, mdr:%d , request: hadr %d, madr %d \n", rplock->ownerHdr,rplock->ownerMdr,request->requestFromAddr,request->requestFromMail);
    if(rplock->ownerHdr==request->requestFromAddr&&request->requestFromMail==rplock->ownerMdr)
        return TRUE;
    else
        return false;
}

bool checkUserValid(RpcReqMessage * request)
{
    return true;
    int i=0;
    for(i=0; i<MAX_USERNUM; i++)
    {
        if((authclients[i].clientHaddr==request->requestFromAddr)&&(authclients[i].clientmail==request->requestFromMail))
        {
           // DEBUGINFO('r', "client %d, mail %d is valid\n", request->requestFromAddr,request->requestFromMail);
            break;
        }
    }
    if(i==MAX_USERNUM)
        return false;
    else
        return true;
}

bool loginOnServer(RpcReqMessage * request)
{
    if(serverinfo->password==request->reqIntParam1)
    {
        //step 1:check if we already set into the table
        if(checkUserValid( request))
        {
            return true;
        }
        else
        {
            if(curAutUsrNum==MAX_USERNUM)
            {
                DEBUGINFO('r', "the auth user number is full, you could not use\n");
                return false;
            }
            else
            {
                authclients[curAutUsrNum].clientHaddr=request->requestFromAddr;
                authclients[curAutUsrNum].clientmail=request->requestFromMail;
                authclients[curAutUsrNum].isValid=true;
                curAutUsrNum++;
                return true;
            }
        }
    }
    else
        return false;
}
void rpcLoginOn(RpcReqMessage * request)
{
    RpcRplyMessage  reply;
    reply.ansType=request->reqType;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    if(loginOnServer(request))
    {
        reply.ansIntParam1=1;
        reply.isValid=true;
    }
    else
    {
        reply.ansIntParam1=-1;
        reply.isValid=false;
    }
    replyRequest(&reply);
}

void rpcCreateLock(RpcReqMessage * request)
{
    int i=0;
    RpcRplyMessage  reply;
    //step 1;check the name valid or not
    if(strlen(request->reqStringParam)>MAX_NAME)
    {
        DEBUGINFO('r', "client %d mail %d try to create a lock with a too long name.\n",request->requestFromAddr,request->requestFromMail);
        i=-1;
    }
    //step 2: to find if there any Lock with the same name
    for(i=0; i<Max_RPC_LOCKNUM; i++)
    {
        if(0==strcmp(request->reqStringParam,rpclock[i].name))
        {
            reply.ansIntParam1=ALREADY_HAVE;
            DEBUGINFO('r', "client %d mail %d try to create a lock named %s, we already have.\n",request->requestFromAddr,request->requestFromMail,request->reqStringParam);
            break;
        }
    }
    //step 3: create the Lock if needed
    if(i==Max_RPC_LOCKNUM)
    {
        i = rpclockTable->Put((void *)request->reqStringParam);
        strcpy(rpclock[i].name,request->reqStringParam);
        rpclock[i].state=RPC_LOCKAVAIABLE;
        rpclock[i].waitList=new List;
        rpclock[i].ownerHdr=-1;
        rpclock[i].ownerMdr=-1;
        rpclock[i].userSum=0;
        rpclock[i].isToDelete=false;
        reply.ansIntParam1=SUCCESS;
        DEBUGINFO('r', "client %d mail %d created a lock %d\n",request->requestFromAddr,request->requestFromMail,i);
    }
    //step 4: reply to the client
    reply.ansType=SC_CreateLock;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.isValid=TRUE;
    replyRequest(&reply);
}

void rpcAcquireLock(RpcReqMessage * request)
{
    int result=0;
    bool needreply=false;
    RpcRplyMessage  reply;
    reply.ansType=SC_AcquireLock;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.ansIntParam1=1;
    reply.pingId=request->pingId;
    reply.isValid=TRUE;
    //step 1;check lock number valid or not
    int i =getLockByName(request->reqStringParam);
    if (i == -1)
    {
        DEBUGINFO('r', "client %d mail %d try to acq %s lock . wrong lock number\n",request->requestFromAddr,request->requestFromMail,request->reqStringParam);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    //step 2: to find the lock is used or not
    //rpclockLock->Acquire();
    //if avaiable, set the owner of the lock
    rpclock[i].userSum++;
    if(RPC_LOCKAVAIABLE== rpclock[i].state)
    {
        DEBUGINFO('r', "client %d mail %d ping %d, acq %s lock .\n",request->requestFromAddr,request->requestFromMail, request->pingId, request->reqStringParam);
        rpclock[i].state=RPC_LOCKINUSE;
        rpclock[i].ownerHdr=request->requestFromAddr;
        rpclock[i].ownerMdr=request->requestFromMail;
        rpclock[i].ownerPingMdr = request->pingId;
        reply.ansIntParam1=1;
        replyRequest(&reply);
        return;
    }
    //else if the lock is inuse ,put the reply in the waitlist, when someone release, reply the msg
    else
    {
        int a=rplyTable->Put((void *)&reply);
        setReply(&(replayArray[a]), &reply);
        rpclock[i].waitList->Append((void *)(&(replayArray[a])));
         DEBUGINFO('r', "client %d mail %d  acq %s lock , but need to wait using reply %d\n",request->requestFromAddr,request->requestFromMail,request->reqStringParam,a);
    }
}

void releaseLock(int i) {
    RpcRplyMessage * replyp;
    rpclockLock->Acquire();
    rpclock[i].userSum--;
    //to check if there are anybody wait for this lock
    if(rpclock[i].waitList->IsEmpty()) //if the waitlist is IsEmpty, just set the state of the lock
    {
        //DEBUGINFO('r',"the %s lock is empty\n" , request->reqStringParam);
        rpclock[i].ownerHdr=-1;
        rpclock[i].ownerMdr=-1;
        rpclock[i].state=RPC_LOCKAVAIABLE;
    }
    //else get a waiting client out of the wait queue,
    else
    {
        //we need get out a reply from the waitlist
        replyp=(RpcRplyMessage *)  rpclock[i].waitList->Remove();
        //DEBUGINFO('r',"hda %d madr %d get the lock %s\n" , replyp->replyToAddr,replyp->replyToMail,request->reqStringParam);
	    DEBUGINFO('r',"hda %d madr %d get the lock %s\n" , replyp->replyToAddr,replyp->replyToMail,rpclock[i].name);
        rpclock[i].ownerHdr=replyp->replyToAddr;
        rpclock[i].ownerMdr=replyp->replyToMail;
        rpclock[i].ownerPingMdr = replyp->pingId;
        rpclock[i].state=RPC_LOCKINUSE;
        replyRequest(replyp);
	 int offset=((int)replyp   - (int) replayArray)/(sizeof(RpcRplyMessage));
        rplyTable->Remove(offset);

    }
    rpclockLock->Release();
}


void rpcReleaseLock(RpcReqMessage * request)
{
    int result=0;
    bool needreply=false;
    RpcRplyMessage  reply;
    RpcRplyMessage * replyp;
    reply.ansType=SC_ReleaseLock;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.isValid=TRUE;
    //step 1;check lock number valid or not, meanwhile we need to check if it the owner of the lock
     int i =getLockByName(request->reqStringParam);
    if((i == -1)||(request->requestFromAddr!=rpclock[i].ownerHdr)||(request->requestFromMail!=rpclock[i].ownerMdr))
    {
        result=-1;
        needreply=true;
        reply.ansIntParam1=result;
        replyRequest(&reply);
        return;
    }
    //step 2: now we, release the lock
    if (request->noNeedReply == false) {
        result=1;
        reply.ansIntParam1=result;
        replyRequest(&reply);
    }

    releaseLock(i);
/*
    rpclockLock->Acquire();
    //to check if there are anybody wait for this lock
    result=1;
    reply.ansIntParam1=result;
    rpclock[i].userSum--;
    if(rpclock[i].waitList->IsEmpty()) //if the waitlist is IsEmpty, just set the state of the lock
    {
        DEBUGINFO('r',"the %s lock is empty\n" , request->reqStringParam);
        rpclock[i].ownerHdr=-1;
        rpclock[i].ownerMdr=-1;
        rpclock[i].state=RPC_LOCKAVAIABLE;
        replyRequest(&reply);
    }
    //else get a waiting client out of the wait queue,
    else
    {
        //step a: we need to reply to the release client at first

        replyRequest(&reply);
        //step b: we need get out a reply from the waitlist
        replyp=(RpcRplyMessage *)  rpclock[i].waitList->Remove();
        DEBUGINFO('r',"hda %d madr %d get the lock %s\n" , replyp->replyToAddr,replyp->replyToMail,request->reqStringParam);
        rpclock[i].ownerHdr=replyp->replyToAddr;
        rpclock[i].ownerMdr=replyp->replyToMail;
        rpclock[i].ownerPingMdr = replyp->pingId;
        rpclock[i].state=RPC_LOCKINUSE;
        replyRequest(replyp);
    }
    rpclockLock->Release();
*/
}
void rpcDeleteLock(RpcReqMessage * request)
{
    int result=0;
    bool needreply=false;
    RpcRplyMessage  reply;
    RpcRplyMessage * replyp;
    reply.ansType=SC_DeleteLock;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.isValid=TRUE;

     int i =getLockByName(request->reqStringParam);
    if (i == -1)
    {
        printf("no such lock\n");
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }

    if (rpclock[i].userSum != 0)
    {
        printf("the lock to delete is in use\n");
        rpclock[i].isToDelete = TRUE;
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    strcpy(rpclock[i].name,"");
    rpclock[i].ownerHdr=-1;
    rpclock[i].ownerMdr=-1;
    rpclock[i].state=RPC_LOCKAVAIABLE;
    rpclock[i].isToDelete = FALSE;
    rpclockTable->Remove(i);
    reply.ansIntParam1=1;
    replyRequest(&reply);
    return;
}

void rpcCreateCV(RpcReqMessage * request)
{
    int i=0;
    RpcRplyMessage  reply;
    reply.ansType=SC_CreateCV;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.isValid=TRUE;
    //step 1;check the name valid or not
    if(strlen(request->reqStringParam)>MAX_NAME)
    {
        DEBUGINFO('r', "client %d mail %d try to create a CV with a too long name.\n",request->requestFromAddr,request->requestFromMail);
        i=-1;
    }
    //step 2: to find if there any CV with the same name
    for(i=0; i<Max_RPC_CVNUM; i++)
    {
        if(0==strcmp(request->reqStringParam,rpcCV[i].name))
        {
            reply.ansIntParam1=ALREADY_HAVE;
            DEBUGINFO('r', "client %d mail %d try to create a lock named %s, we already have.\n",request->requestFromAddr,request->requestFromMail,request->reqStringParam);
            break;
        }
    }
    //step 3: create the CV if needed
    if(i==Max_RPC_CVNUM)
    {
        i = rpcCVTable->Put((void*)request->reqStringParam);
        strcpy(rpcCV[i].name,request->reqStringParam);
        rpcCV[i].waitinglockid=-1;
        rpcCV[i].waitList=new List;
        rpcCV[i].userSum=0;
        rpcCV[i].isToDelete=false;
        reply.ansIntParam1=SUCCESS;
        DEBUGINFO('r', "client %d mail %d created a CV %s using index %d\n",request->requestFromAddr,request->requestFromMail,request->reqStringParam,i);
    }
    //step 4: reply to the client
    replyRequest(&reply);
}

void rpcWaitCV(RpcReqMessage * request)
{
    int result =0;
    RpcRplyMessage  reply;
    reply.ansType=SC_WaitCV;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.pingId = request->pingId;
    reply.isValid=TRUE;
    //step 1: check lockid valid
    int i = request->waitinglockid;
/*    int i =getLockByName(request->reqStringParam);
    if (i == -1)
    {
        DEBUGINFO('r', "client %d mail %d wait for a wrong lock %d\n",request->requestFromAddr,request->requestFromMail,i);
        result= -1;
        reply.ansIntParam1=result;
        replyRequest(&reply);
        return;
    }
*/
    //step 2: check cvid valid
    int j = getCVByName(request->reqStringParam1);
    if (j == -1)
    {
        DEBUGINFO('r', "client %d mail %d wait for a wrong CV %d\n",request->requestFromAddr,request->requestFromMail,j);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    //step 3: check the lock is the waiting lock
    // TODO had better change id to char[] 
    if (-1==rpcCV[j].waitinglockid)
    {
        DEBUGINFO('r', "client %d mail %d wait for lock %d the first time\n",request->requestFromAddr,request->requestFromMail,i);
        rpcCV[j].waitinglockid=i;
    }
    //step 3: check the lock is the waiting lock
// TODO: Need to add return info in DYH reply
    if (i!=rpcCV[j].waitinglockid)
    {
        DEBUGINFO('r', "client %d mail %d wait for lock %d, but it is the owner\n",request->requestFromAddr,request->requestFromMail,i);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    //step 4: check it is the owner of the lock
/*    if (!isOwner(&rpclock[i],request))
    {
        DEBUGINFO('r', "client %d mail %d wait for lock %d, but it is the owner\n",request->requestFromAddr,request->requestFromMail,i);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
*/
    //step 5: all passed, do the wait operation
    DEBUGINFO('r', "client %d mail %d  is added to cv %d waitlist\n",request->requestFromAddr,request->requestFromMail,j);
    //step a: if waitlinglock is non-iniated, set it
    if(-1==rpcCV[j].waitinglockid)
        rpcCV[j].waitinglockid=i;
    //step b: we need to release the lock and make anyother who want to get the lock and signal do it
    int a=rplyTable->Put((void *)&reply);
    setReply(&(replayArray[a]), &reply);
    rpcCV[j].waitList->Append((void *)(&(replayArray[a])));
    DEBUGINFO('r', "client %d mail %d  is added to cv %d waitlist using %d reply\n",reply.replyToAddr,reply.replyToMail,j,a);
    rpcCV[j].userSum++;

    request->reqType = SC_ReleaseLock;
    request->noNeedReply = true;
    forwardRequest(request, request->lockServerId, 0);
/*    if(rpclock[i].waitList->IsEmpty())
    {
        rpclock[i].ownerHdr=-1;
        rpclock[i].ownerMdr=-1;
        rpclock[i].state=RPC_LOCKAVAIABLE;
    }
    else
    {
        RpcRplyMessage *  replyp=(RpcRplyMessage *)  rpclock[i].waitList->Remove();
        rpclock[i].ownerHdr=replyp->replyToAddr;
        rpclock[i].ownerMdr=replyp->replyToMail;
        rpclock[i].state=RPC_LOCKINUSE;
        replyRequest(replyp);
    }
*/
}
void rpcSignalCV(RpcReqMessage * request)
{
    int result =0;
    RpcRplyMessage  reply;
    reply.ansType=SC_SignalCV;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.isValid=TRUE;
    //step 1: check lockid valid
     int i = request->waitinglockid;//getLockByName(request->reqStringParam);
/*    if (i == -1)
    {
        DEBUGINFO('r', "client %d mail %d signal a wrong lock %d\n",request->requestFromAddr,request->requestFromMail,i);
        result= -1;
        reply.ansIntParam1=result;
        replyRequest(&reply);
        return;
    }
*/
    //step 2: check cvid valid
   int j = getCVByName(request->reqStringParam1);
    if (j== -1)
    {
        DEBUGINFO('r', "client %d mail %d signal a wrong CV %d\n",request->requestFromAddr,request->requestFromMail,j);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    //step 3.1: check it is presignal
    if (-1==rpcCV[j].waitinglockid)
    {
        DEBUGINFO('r', "client %d mail %d signal for lock %d, but presignal\n",request->requestFromAddr,request->requestFromMail,request->reqIntParam1);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    //step 3: check the lock is the waiting lock
    if (i!=rpcCV[j].waitinglockid)
    {
        DEBUGINFO('r', "client %d mail %d signal lock %d, but it is not the waitinglock\n",request->requestFromAddr,request->requestFromMail,request->reqIntParam1);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    /*step 4: check it is the owner of the lock
    if (!isOwner(&rpclock[i],request))
    {
        DEBUGINFO('r', "client %d mail %d signal lock %d, but it is not the owner\n",request->requestFromAddr,request->requestFromMail,request->reqIntParam1);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }*/
    //step 5: all passed, get a reply from the waitlist and signal it
    //step 5.a: if the waitlist is IsEmpty, return -1
    if(rpcCV[j].waitList->IsEmpty())
    {
        DEBUGINFO('r', "client %d mail %d signal cv %d, but it is the IsEmpty\n",request->requestFromAddr,request->requestFromMail,j);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    //tep 5.b: copy a one from cv waitlist to the lock waitlist, so then release, the wait client could run
    RpcRplyMessage * rplyp ;
    int offset=0;
    rplyp=(RpcRplyMessage *) rpcCV[j].waitList->Remove();
    offset=((int)rplyp   - (int) replayArray)/(sizeof(RpcRplyMessage));
    DEBUGINFO('r', "rplyp %d, rpccv %d ,rplypp %d  sizeof %d offset %d \n", rplyp,replayArray,sizeof(RpcRplyMessage),offset);
    //step 5.c: set the wait client to the lock-waitlist, so when the signal client try to release lock, it could run
    DEBUGINFO('r', "client %d mail %d was put from signal cv %s to lock %s waitlist,\n", rplyp->replyToAddr,rplyp->replyToMail,rpcCV[j].name,rpclock[i].name);
    rpcCV[j].userSum--;

    request->requestFromAddr = rplyp->replyToAddr;
    request->requestFromMail = rplyp->replyToMail;
    request->pingId = rplyp->pingId;
    request->reqType = SC_AcquireLock;
    request->noNeedReply = true;
    forwardRequest(request, request->lockServerId, 0);
    //rpclock[i].waitList->Append((void *) rplyp);
    //step 6 reply to the client, we do it succes
    reply.ansIntParam1=1;
    replyRequest(&reply);
    rplyTable->Remove(offset); 
}

void rpcBroadcastCV(RpcReqMessage * request)
{
    int result =0;
    RpcRplyMessage  reply;
    reply.ansType=SC_BroadcastCV;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.isValid=TRUE;
    //step 1: check lockid valid
    int i = request->waitinglockid;  // getLockByName(request->reqStringParam);
/*    if (i == -1)
    {
        DEBUGINFO('r', "client %d mail %d broadcast for a wrong lock %d\n",request->requestFromAddr,request->requestFromMail,request->reqIntParam1);
        result= -1;
        reply.ansIntParam1=result;
        replyRequest(&reply);
        return;
    }
*/
    //step 2: check cvid valid
    int j =getCVByName(request->reqStringParam1);
    if (j == -1)
    {
        DEBUGINFO('r', "client %d mail %d broadcast for a wrong CV %d\n",request->requestFromAddr,request->requestFromMail,j);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    //step 3.1: check it is presignal
    if (-1==rpcCV[j].waitinglockid)
    {
        DEBUGINFO('r', "client %d mail %d c for lock %d, but presignal\n",request->requestFromAddr,request->requestFromMail,i);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    //step 3: check the lock is the waiting lock
    if (i!=rpcCV[j].waitinglockid)
    {
        DEBUGINFO('r', "client %d mail %d broadcast for lock %d, but it is not the waiting lock\n",request->requestFromAddr,request->requestFromMail,i);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    //step 4: check it is the owner of the lock
/*    if (!isOwner(&rpclock[i],request))
    {
        DEBUGINFO('r', "client %d mail %d broadcast for lock %d, but it is the owner\n",request->requestFromAddr,request->requestFromMail,i);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
*/
    //step 5: all passed, get a reply from the waitlist and signal it
    //tep 5.a: get all from cv waitlist to the lock waitlist, so then release, the wait client could run
    RpcRplyMessage * rplyp ;
    while(!rpcCV[j].waitList->IsEmpty())
    {
        //step 5.b: set the wait client to the lock-waitlist, so when the signal client try to release lock, it could run
        rplyp=(RpcRplyMessage *) rpcCV[j].waitList->Remove();
        DEBUGINFO('r', "client %d mail %d was put from signal cv %s to lock %s waitlist,\n", rplyp->replyToAddr,rplyp->replyToMail,rpcCV[j].name,rpclock[i].name);
        request->requestFromAddr = rplyp->replyToAddr;
        request->requestFromMail = rplyp->replyToMail;
        request->pingId = rplyp->pingId;
        request->reqType = SC_AcquireLock;
        request->noNeedReply = true;
        forwardRequest(request, request->lockServerId, 0);

//        rpclock[i].waitList->Append(rplyp);
    }
    //step 6 reply to the client, we do it succes
    reply.ansIntParam1=1;
    replyRequest(&reply);
}

void rpcDeleteCV(RpcReqMessage * request)
{
    int result=0;
    bool needreply=false;
    RpcRplyMessage  reply;
    RpcRplyMessage * replyp;
    reply.ansType=SC_DeleteCV;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.isValid=TRUE;

    int i=getCVByName(request->reqStringParam);

    if (i== -1)
    {
        printf("no such CV\n");
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }

    if (rpcCV[i].userSum != 0)
    {
        printf("the CV to delete is in use\n");
        rpcCV[i].isToDelete = TRUE;
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    strcpy(rpcCV[i].name,"");
    rpcCV[i].waitinglockid=-1;
    rpcCV[i].isToDelete = FALSE;
    rpcCVTable->Remove(i);
    reply.ansIntParam1=1;
    replyRequest(&reply);
}

void rpcCreateMV(RpcReqMessage * request)
{
    int i=0;
    RpcRplyMessage  reply;
    reply.ansType=SCRPC_CreateMV;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.isValid=TRUE;
    //step 1;check the name valid or not
    if(strlen(request->reqStringParam)>MAX_NAME)
    {
        DEBUGINFO('r', "client %d mail %d try to create a MV with a too long name.\n",request->requestFromAddr,request->requestFromMail);
        i=-1;
    }
    //step 2: to find if there any CV with the same name
    for(i=0; i<Max_RPC_MVNUM; i++)
    {
        if(0==strcmp(request->reqStringParam,rpcMV[i].name))
        {
            DEBUGINFO('r', "client %d mail %d try to create a lock named %s, we already have.\n",request->requestFromAddr,request->requestFromMail,request->reqStringParam);
            reply.ansIntParam1=ALREADY_HAVE;
            break;
        }
    }
    //step 3: create the CV if needed
    if(i==Max_RPC_MVNUM)
    {
        i = rpcMVTable->Put((void*)1);
        strcpy(rpcMV[i].name,request->reqStringParam);
        rpcMV[i].value=0;
        reply.ansIntParam1=SUCCESS;
        DEBUGINFO('r', "client %d mail %d created a MV %d  name %s\n",request->requestFromAddr,request->requestFromMail,i,rpcMV[i].name);
    }
    //step 4: reply to the client
    replyRequest(&reply);
}

void rpcGetMV(RpcReqMessage * request)
{
    RpcRplyMessage  reply;
    reply.ansType=SCRPC_GetMV;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.isValid=TRUE;
    //step 1;check the name valid or not
    int i =getMVByName(request->reqStringParam);
    if (i == -1)
    {
        DEBUGINFO('r', "client %d mail %d try to get % mv . wrong mv number\n",request->requestFromAddr,request->requestFromMail,i);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    //step 2: return the mv's value
 
    reply.ansIntParam1=rpcMV[i].value;
    replyRequest(&reply);
}

void rpcSetMV(RpcReqMessage * request)
{
    RpcRplyMessage  reply;
    reply.ansType=SCRPC_SetMV;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.isValid=TRUE;
    //step 1;check the name valid or not
    int i =getMVByName(request->reqStringParam);
    if (i == -1)
    {
        DEBUGINFO('r', "client %d mail %d try to set %s mv . wrong mv number\n",request->requestFromAddr,request->requestFromMail,request->reqStringParam);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }
    //step 2: return the mv's value
    rpcMV[i].value=request->reqIntParam2;
    reply.ansIntParam1=1;
    replyRequest(&reply);
}
void rpcDeleteMV(RpcReqMessage * request){


    RpcRplyMessage  reply;
    reply.ansType=SCRPC_DeleteMV;
    reply.replyToAddr=request->requestFromAddr;
    reply.replyToMail=request->requestFromMail;
    reply.isValid=TRUE;
    //step 1;check the name valid or not

    int i =getMVByName(request->reqStringParam);
    if (i == -1)
    {
        DEBUGINFO('r', "client %d mail %d try to delete %s mv . wrong mv number\n",request->requestFromAddr,request->requestFromMail,request->reqStringParam);
        reply.ansIntParam1=-1;
        replyRequest(&reply);
        return;
    }

    
    //step 2: return the mv's value
   
    strcpy(rpcMV[i].name,"");
    rpcMVTable->Remove(i);
    reply.ansIntParam1=1;
    replyRequest(&reply);


}

void checkDeleteTask(int arg)
{
    int i=0;
    for (i = 0; i < Max_RPC_LOCKNUM; ++i)
    {
        void * id = rpclockTable->Get(i);
        if (int(id) == 0)
        {
            continue;
        }
        if (!rpclock[i].userSum && rpclock[i].isToDelete == TRUE)
        {
            strcpy(rpclock[i].name,"");
            rpclock[i].ownerHdr=-1;
            rpclock[i].ownerMdr=-1;
            rpclock[i].state=RPC_LOCKAVAIABLE;
            rpclock[i].isToDelete = FALSE;
            rpclockTable->Remove(i);
            DEBUGINFO('r', "delete the lock %d in deletetask\n", i);
        }
    }
    for (i = 0; i < Max_RPC_CVNUM; ++i)
    {
        void * id = rpcCVTable->Get(i);
        if (int(id) == 0)
        {
            continue;
        }
        if (!rpcCV[i].userSum && rpcCV[i].isToDelete == TRUE)
        {
            strcpy(rpcCV[i].name,"");
            rpcCV[i].waitinglockid=-1;
            rpcCV[i].isToDelete = FALSE;
            rpcCVTable->Remove(i);
            DEBUGINFO('r', "delete the cv %d in deletetask\n", i);
        }
    }


}

int lastTime = 0;
void serverPingHandler(int arg) {
  int currentTime = time(NULL);
  if (currentTime<lastTime+15) {
    return;
  }
  lastTime = currentTime;
  ServerInfo * ping = new ServerInfo;
  // TODO serverHaddr
  //ping->serverHaddr = netname;
  ping->serverHaddr = currentThread->haddr;
  ping->serverMail = currentThread->mailId;
  int i=0;

  RpcReqMessage rpcReq;
  //rpcReq.reqIntParam1=password;
  //while (1) {
    //for (i = 0;i < 1000000; ++i) {
    //  currentThread->Yield();
    //}
    for (i = 0; i < Max_RPC_LOCKNUM; ++i)
    {
        void * id = rpclockTable->Get(i);
        if (int(id) == 0)
        {
            continue;
        }
        if (rpclock[i].state == RPC_LOCKAVAIABLE) continue;

        DEBUGINFO('P', "Ping owner of Lock %s %d , client hdr %d, box %d\n", rpclock[i].name, i, rpclock[i].ownerHdr, rpclock[i].ownerPingMdr);
        // ping lock owner
        if (!sendRequest(&rpcReq, rpclock[i].ownerHdr, rpclock[i].ownerPingMdr)) {
            DEBUGINFO('P', "Owner of Lock %s %d is dead, client hdr %d, box %d\n", rpclock[i].name, i, rpclock[i].ownerHdr, rpclock[i].ownerPingMdr);
            // TODO: if lock owner is dead, release the lock
            releaseLock(i);
/*
            bzero(rpcReq.reqStringParam,MAXREQSTRING);
            strcpy(rpcReq.reqStringParam, rpclock[i].name);
    
            rpcReq.reqType=SC_ReleaseLock;
            rpcReq.reqIntParam1 = strlen(rpclock[i].name);
            sendRequest(&rpcReq, currentThread->haddr, 0);
*/
        }
    }
  //}
  }
}

void distributeRPCReq(RpcReqMessage * rpcreq)
{
    int type=rpcreq->reqType;
    DEBUGINFO('n', "request type %d\n",type);
    switch (type)
    {
    case SC_CreateLock:
        DEBUGINFO('n', "rpc Create lock syscall.\n");
        rpcCreateLock(rpcreq);
        break;
    case SC_AcquireLock:
        DEBUGINFO('n', "rpc rpcAcqLock syscall.\n");
        rpcAcquireLock(rpcreq);
        break;
    case SC_ReleaseLock:
        DEBUGINFO('n', "rpc rpcReleaseLock syscall.\n");
        rpcReleaseLock(rpcreq);
        break;
    case SC_CreateCV:
        DEBUGINFO('n', "rpc rpcCreateCV syscall.\n");
        rpcCreateCV(rpcreq);
        break;
    case SC_WaitCV:
        DEBUGINFO('n', "rpc RPcWaitCV syscall.\n");
        rpcWaitCV(rpcreq);
        break;
    case SC_SignalCV:
        DEBUGINFO('n', "rpc rpcSignalCV syscall.\n");
        rpcSignalCV(rpcreq);
        break;
    case SC_BroadcastCV:
        DEBUGINFO('n', "rpc SCRPC_BroadcastCV syscall.\n");
        rpcBroadcastCV(rpcreq);
        break;
    case SC_DeleteCV:
        DEBUGINFO('n', "rpc SCRPC_deleteCV syscall.\n");
        rpcDeleteCV(rpcreq);
        break;
    case SC_DeleteLock:
        DEBUGINFO('n', "rpc SCRPC_deletelock syscall.\n");
        rpcDeleteLock(rpcreq);
        break;
    case SCRPC_CreateMV:
        DEBUGINFO('n', "rpc SCRPC_BroadcastCV syscall.\n");
        rpcCreateMV(rpcreq);
        break;
    case SCRPC_GetMV:
        DEBUGINFO('n', "rpc SCRPC_deleteCV syscall.\n");
        rpcGetMV(rpcreq);
        break;
    case SCRPC_SetMV:
        DEBUGINFO('n', "rpc SCRPC_deletelock syscall.\n");
        rpcSetMV(rpcreq);
        break;
    case SCRPC_DeleteMV:
        DEBUGINFO('n', "rpc SCRPC_deletemv syscall.\n");
        rpcDeleteMV(rpcreq);
   break;

    
    default:
        DEBUGINFO('n', "Unknown syscall - shutting down.\n");
    }
}
int searchEntityTable(int type,char *name){

   
    int ret=-1;
    int i=0;
    switch (type)
    {
    case SC_CreateLock:
    case SC_AcquireLock:
    case SC_ReleaseLock:
    case SC_DeleteLock:
    case SC_CV2:
         for(i=0; i<Max_RPC_LOCKNUM; i++)
        {
            if(rpclockTable->Get(i)!=0){
                if(0==strcmp(name,rpclock[i].name))
                {
                    ret=i;
                    break;
                }
            }
        }
    break;
    case SC_CreateCV:
    case SC_DeleteCV:
    case SC_WaitCV:
    case SC_SignalCV:
    case SC_BroadcastCV:
        for(i=0; i<Max_RPC_CVNUM; i++)
        {
            if(rpcCVTable->Get(i)!=0){
                if(0==strcmp(name,rpcCV[i].name))
                {
                    ret=i;
                    break;
                }
            }
        }
    break;
    case SCRPC_CreateMV:
    case SCRPC_GetMV:
    case SCRPC_SetMV:
        for(i=0; i<Max_RPC_MVNUM; i++)
        {

             if(rpcMVTable->Get(i)!=0){
                if(0==strcmp(name,rpcMV[i].name))
                {
                    ret=i;
                    break;
                }
            }
        }
    break; 
    }

    return ret;

}

int sendDYHRequest(RpcReqMessage * rpcreq, bool isCVForLock = false){


    PacketHeader outPktHdr;
    MailHeader outMailHdr;
    char databuffer[MaxMailSize];
    bool success=false;


    RpcDYHMessage *DYHMsg;
    DYHMsg=new RpcDYHMessage();
    

    DYHMsg->type=rpcreq->reqType;
    
    if((DYHMsg->type==SC_WaitCV
      ||DYHMsg->type==SC_SignalCV||DYHMsg->type==SC_BroadcastCV) && !isCVForLock)
    {   // cv name is in param1
        strcpy(DYHMsg->name,rpcreq->reqStringParam1);                                      
    }
    else
    {                        
        strcpy(DYHMsg->name,rpcreq->reqStringParam);                         
    }
    DYHMsg->isCVForLock = isCVForLock;

    
   //search if there is same request on the same server
    if(DYHMsg->type==SC_CreateLock||DYHMsg->type==SC_CreateCV||DYHMsg->type==SCRPC_CreateMV || 
        DYHMsg->type==SC_DeleteLock||DYHMsg->type==SC_DeleteCV||DYHMsg->type==SCRPC_DeleteMV)
    {
        for(int i=0;i<MAX_PENDING_REQUEST;i++){
            if(prTable->Get(i) && strcmp(DYHMsg->name,pendingRequest[i].name)==0 && DYHMsg->type == pendingRequest[i].type)
            {            
                DEBUGINFO('r', "append to pending request[%d] %s, %d, %d %d, req m %d, box %d \n", i, DYHMsg->name, DYHMsg->type, pendingRequest[i].clientAddr, pendingRequest[i].clientMail, rpcreq->requestFromAddr, rpcreq->requestFromMail);
                if (pendingRequest[i].sameReq == NULL) {
                    pendingRequest[i].sameReq = new List();
                }
                int a= reqTable->Put((void *)rpcreq);
                setReq(&(reqArray[a]), rpcreq);
                pendingRequest[i].sameReq->Append((void*)(&(reqArray[a]))); 
/*
                ListElement * it = pendingRequest[i].sameReq->first;
                while (true) {
                    RpcReqMessage * req = (RpcReqMessage*) it->item; 
                    printf("tttttttttttt requestFromMail %d\n", req->requestFromMail);
                    if (it == pendingRequest[i].sameReq->last) {
                        break;
                    }
                    it = it->next;
                }
*/

                return success;
            }        

        }
    }

    DYHMsg->timestamp=time(NULL);
    DEBUGINFO('r', "time stamp [%d]\n",DYHMsg->timestamp);
    DYHMsg->msgtype=DYH_REQUEST;
    DYHMsg->reply=DYH_REPLY_INIT;
    





    //Add rpc req entry to pending request table
    int index=prTable->Put((void *)rpcreq);      
    strcpy(pendingRequest[index].name,DYHMsg->name);
    pendingRequest[index].type=DYHMsg->type;
    pendingRequest[index].replyNoNumber=1;  
    pendingRequest[index].clientAddr=rpcreq->requestFromAddr;
    pendingRequest[index].clientMail=rpcreq->requestFromMail;
    pendingRequest[index].timestamp=DYHMsg->timestamp;
    memcpy(&pendingRequest[index].orgClientReq,rpcreq, sizeof(RpcReqMessage));
  
    
    //TODO: decide destination
    //send DYH message to all server
    for(int i=0;i<serverNumber;i++){

        if(currentThread->haddr!=serversAddr[i]){
            outPktHdr.from=currentThread->haddr;
            outPktHdr.to = serversAddr[i];
            outMailHdr.to = 1;
            outMailHdr.from = 1;
            outMailHdr.length=sizeof(RpcDYHMessage);
            bcopy( (char *) DYHMsg, databuffer, sizeof(RpcDYHMessage));
            success = postOffice->Send(outPktHdr, outMailHdr,databuffer);
        }
    }

     //TODO:
    //if server number is one,send no reply to my self or check number of no immediately
    //because no server will reply.
    if(serverNumber==1){
        outPktHdr.from=netname;
        outPktHdr.to = netname;
        outMailHdr.to = currentThread->mailId;
        outMailHdr.from = currentThread->mailId;
        outMailHdr.length=sizeof(RpcDYHMessage);
        DYHMsg->msgtype=DYH_REPLY;
        DYHMsg->reply=DYH_REPLY_NO;
        bcopy( (char *) DYHMsg, databuffer, sizeof(RpcDYHMessage));
        success = postOffice->Send(outPktHdr, outMailHdr,databuffer);
    }


    delete DYHMsg;
    return success;


}
int replyDYH(RpcDYHMessage * dyh,int reply){


    PacketHeader outPktHdr;
    MailHeader outMailHdr;
    char databuffer[MaxMailSize];
    
    dyh->reply=reply;
    dyh->msgtype=DYH_REPLY;
    
    outPktHdr.from=currentThread->haddr;
    outPktHdr.to = dyh->FromAddr;
    outMailHdr.to = dyh->FromMail;
    outMailHdr.from = currentThread->mailId;
    outMailHdr.length=sizeof(RpcDYHMessage);
    bcopy( (char *) dyh, databuffer, sizeof(RpcDYHMessage));
    DEBUGINFO('r',"replyDYH to m %d, to %d  msgtype: %d   aaaaaaaaaaaaaaaaaaaaa", dyh->FromAddr, dyh->FromMail, dyh->msgtype);
    bool success = postOffice->Send(outPktHdr, outMailHdr,databuffer);
    
    return success;


}
int searchPRTable(int type, char *name){

    int ret=-1;

    for(int i=0;i<MAX_PENDING_REQUEST;i++){
        if(prTable->Get(i)!=0){
            if(strcmp(pendingRequest[i].name,name)==0 && 
                type==pendingRequest[i].type){
                ret=i;
            }    
        }
    }
    return ret;

}
void DYHReceive(int i){

    int found=-1;

    RpcDYHMessage *dyh = new RpcDYHMessage();
    while(true){

        //wait for the DYH reply/request from other server

        waitForDYH(dyh);
        if(dyh->msgtype==DYH_REPLY){//1.reveive a DYH reply 

            //2.extract name and type
            //3.look for matching entry in pending request table  
            found=searchPRTable(dyh->type,dyh->name);            

            if(found==-1)
            {
                //4.if not found(already has a yes), just return   
                continue;
            }else{  //5.if found in pending request table,     

                if(dyh->reply==DYH_REPLY_YES){  //5.1 if reply is yes                                            
                    if(dyh->type==SC_CreateLock||dyh->type==SC_CreateCV||dyh->type==SCRPC_CreateMV)
                    {
                        //5.1.1 if create type,send reply to client success        

                        RpcRplyMessage  reply;
                        reply.ansType=dyh->type;
                        reply.replyToAddr=pendingRequest[found].orgClientReq.requestFromAddr;
                        reply.replyToMail=pendingRequest[found].orgClientReq.requestFromMail;
                        reply.isValid=TRUE;
                        reply.ansIntParam1=SUCCESS;
                        replyRequest(&reply);
                        while(pendingRequest[found].sameReq != NULL && !pendingRequest[found].sameReq->IsEmpty())
                        {
                            RpcReqMessage * request = (RpcReqMessage *) pendingRequest[found].sameReq->Remove();
                            int offset=((int)request   - (int) reqArray)/(sizeof(RpcReqMessage));
                            reqTable->Remove(offset);
                            reply.replyToAddr = request->requestFromAddr;
                            reply.replyToMail = request->requestFromMail;

                            replyRequest(&reply);
                        }
                    }else{
                        //5.1.2 not create type, forward to the right server
                        if ((dyh->type == SC_WaitCV||dyh->type==SC_SignalCV||dyh->type==SC_BroadcastCV) && dyh->isCVForLock) {
                            pendingRequest[found].orgClientReq.lockServerId = dyh->FromAddr;
                            pendingRequest[found].orgClientReq.lockMailId = dyh->FromMail;
                            pendingRequest[found].orgClientReq.waitinglockid = dyh->lockId;
                            distributeRPCReq(&pendingRequest[found].orgClientReq);
                        } else { 
                            forwardRequest(&pendingRequest[found].orgClientReq, dyh->FromAddr, 0);
                        }
                    } 
                    //5.1.3 remove pending request entry
                    prTable->Remove(found); 
                }else{ //5.2 if  reply is no
                    //5.2.1 increament number of no replies
                    pendingRequest[found].replyNoNumber++;
                    //5.2.2 if number of no replies is equal to number of servers
                    if(pendingRequest[found].replyNoNumber>=serverNumber)
                    {
                        if(dyh->type==SC_CreateLock||dyh->type==SC_CreateCV||dyh->type==SCRPC_CreateMV)
                        {
                            //5.2.2.1 if create type,create and send client reply, 
                            //5.2.2.1 if pending DYH message is not empty, reply all of them
                            

                            distributeRPCReq(&pendingRequest[found].orgClientReq);
                            while(pendingRequest[found].pendingDYHmsg != NULL && !pendingRequest[found].pendingDYHmsg->IsEmpty())
                            {
                                RpcDYHMessage * dyhRequest=(RpcDYHMessage *) pendingRequest[found].pendingDYHmsg->Remove();
                                replyDYH(dyhRequest,DYH_REPLY_YES);          

                                int offset=((int)dyhRequest   - (int) dyhArray)/(sizeof(RpcDYHMessage));
                                dyhTable->Remove(offset);

                            }
                            //reply the same request to the client         
                            while(pendingRequest[found].sameReq != NULL && !pendingRequest[found].sameReq->IsEmpty())
                            {
                                RpcReqMessage * request = (RpcReqMessage *) pendingRequest[found].sameReq->Remove();

                                RpcRplyMessage  reply;
                                reply.ansType=request->reqType;
                                reply.replyToAddr=request->requestFromAddr;
                                reply.replyToMail=request->requestFromMail;
                                reply.isValid=TRUE;
                                reply.ansIntParam1=SUCCESS;
                                int offset=((int)request   - (int) reqArray)/(sizeof(RpcReqMessage));
                                reqTable->Remove(offset);
                                replyRequest(&reply);
                            }

                        }else{

                            // if not create type, do nothing. wait for yes reply
                            continue;
                            //5.2.2.2 if not create type,send an error msg to client 

                            RpcRplyMessage  reply;
                            reply.ansType=dyh->type;
                            reply.replyToAddr=pendingRequest[found].orgClientReq.requestFromAddr;
                            reply.replyToMail=pendingRequest[found].orgClientReq.requestFromMail;
                            reply.isValid=TRUE;
                            reply.ansIntParam1=-1;
                            replyRequest(&reply);

                        } 
                        //5.1.3 remove pending request entry
                        prTable->Remove(found); 
                    }else{ //5.2.3 not equal to number of servers

                        //5.2.3.1 I don't have all no replies, I already increase NO reply. I can't process the msg until I have all no replies                

                    }                                    
                }  
            }                
        }else{
            //1.receive a DYH request
            //2.extract name and type      
            //3.look for matching entry in entry(lock/mv/cv) table 
            int rv = searchEntityTable(dyh->type,dyh->name);
            DEBUGINFO('r', "receive a DYH request found : %d ", rv); 
            //if(searchEntityTable(dyh->type,dyh->name)==0)
            if (rv != -1)
            { //3.1 if found send yes reply
                if (dyh->type == SC_AcquireLock || dyh->type == SC_ReleaseLock) {
                    dyh->lockId = rv; 
                }        
                replyDYH(dyh,DYH_REPLY_YES);          

            }else{ //3.2 if not found

                if(dyh->type==SC_CreateLock||dyh->type==SC_CreateCV||dyh->type==SCRPC_CreateMV)
                {
                    //3.2.1 if create type
                    //3.2.1.1 search pending request table
                    found=searchPRTable(dyh->type,dyh->name);   
                    if(found==-1)
                    {
                        //3.2.1.2 if not found, send no reply

                        replyDYH(dyh,DYH_REPLY_NO);        

                    }else{//3.2.1.3 if found in pending request table

                        if(dyh->timestamp>pendingRequest[found].timestamp){
                            //3.2.1.3.1 if my timestamp & machine ID is later, send no reply

                            replyDYH(dyh,DYH_REPLY_NO);                                 

                        }else if(dyh->timestamp==pendingRequest[found].timestamp && dyh->FromAddr < currentThread->haddr){
                            //3.2.1.3.1 if my timestamp & machine ID is later, send no reply
                            //3.2.1.3.1 if pending DYH message is not empty, reply all of them

                            replyDYH(dyh,DYH_REPLY_NO);   
                        }else{
                            //3.2.1.3.2 if my timestamp is early, queue DYH msg, reply those who send me request later                     
                            if (pendingRequest[found].pendingDYHmsg == NULL) {
                                pendingRequest[found].pendingDYHmsg = new List();
                            }

                            int a=dyhTable->Put((void *)dyh);
                            setDYH(&(dyhArray[a]), dyh);

                            pendingRequest[found].pendingDYHmsg->Append((void *)(&(replayArray[a]))); 

                        }                                                        
                    }

                }else{
                    //3.2.2 if not create type, send no

                    replyDYH(dyh,DYH_REPLY_NO);  
                } 

            }
        }
    }//while
                                                                                    
}

void
rpcServer(int serverId)
{
    char buffer[MaxMailSize];
    PacketHeader outPktHdr;
    MailHeader outMailHdr;
    RpcRplyMessage rpcAnsmsg;
    int reqType;
    char reqName[MAXREQSTRING];

    currentThread->haddr = netname;
    currentThread->mailId = 0;
    Thread * rpcdeamon= new Thread("rpcdeamon");
   // rpcdeamon->Fork(checkDeleteTask, 0);
    Thread * rpcDYHReceive= new Thread("rpcDYHReceive");
    rpcDYHReceive->mailId = 1;
    rpcDYHReceive->haddr = netname;
    rpcDYHReceive->Fork(DYHReceive, 0);
    /*Thread * rpcPing = new Thread("rpcPing");
    rpcPing->mailId = 2;
    rpcPing->haddr = netname;
    rpcPing->Fork(serverPing, 0); 
*/
    timer = new Timer(serverPingHandler, 0, 0);
    if (1)
    {
        while(1)
        {
            RpcReqMessage rpcp;
            //step 1: wait for the request of the client
            waitForRequst(&rpcp);
            //step 2: validate the use
            if(SCRPC_LOGINON==rpcp.reqType)
            {
                rpcLoginOn(&rpcp);
            }
            else if(checkUserValid(&rpcp))
            {
                if(SCRPC_CloseSever==rpcp.reqType)
                {
                    printf("client %d, mail %d close the server\n",rpcp.requestFromAddr,rpcp.requestFromMail);
                    RpcRplyMessage  reply;
                    reply.ansType=rpcp.reqType;
                    reply.replyToAddr=rpcp.requestFromAddr;
                    reply.replyToMail=rpcp.requestFromMail;
                    reply.ansIntParam1=1;
                    reply.isValid=true;
                    replyRequest(&reply);
                    goto out;
                }
                else{
                    //search my table for the lock/cv/mv entity  
                                       
                    reqType=rpcp.reqType;
                    DEBUGINFO('r', "rpcp.reqStringParam %s, rpcp.reqStringParam1 %s", rpcp.reqStringParam, rpcp.reqStringParam1);
                    if(reqType==SC_WaitCV ||reqType==SC_SignalCV||reqType==SC_BroadcastCV)
                    {   // cv name is in param1
                        strcpy(reqName,rpcp.reqStringParam1);                      
                    
                    }else{                        
                        strcpy(reqName,rpcp.reqStringParam);                         
                    }

                   
                    if(serverNumber == 1 || searchEntityTable(reqType,reqName)!= -1){
                        DEBUGINFO('r', "lock/cv/mv belongs to me, do next, reqType %d,reqName %s, serverNumber %d", reqType, reqName, serverNumber);
                    //if lock/cv/mv belongs to me 
                        if (reqType==SC_WaitCV || reqType==SC_SignalCV||reqType==SC_BroadcastCV) {
                            int rv = searchEntityTable(SC_CV2, rpcp.reqStringParam);
                            if (rv == -1) {
                                sendDYHRequest(&rpcp, true);
                            } else {
                                rpcp.lockServerId = currentThread->haddr;
                                rpcp.lockMailId = currentThread->mailId;
                                rpcp.waitinglockid = rv;
                                distributeRPCReq(&rpcp);
                            }
                        } else {
                            distributeRPCReq(&rpcp);
                        }
                    }else{
                     //if not found, 
                     //a. send DYH to all the other servers
                     //b. add entry to pending request table 
                     //note:if request wait,signal,broadcast,send cv and lockid                  
                        DEBUGINFO('r', "lock/cv/mv don't belongs to me, send DYH");
                        sendDYHRequest(&rpcp);
                    }
                }
            }
            else
            {
                printf("client %d, mail %d unauth\n",rpcp.requestFromAddr,rpcp.requestFromMail);
                RpcRplyMessage  reply;
                reply.ansType=rpcp.reqType;
                reply.replyToAddr=rpcp.requestFromAddr;
                reply.replyToMail=rpcp.requestFromMail;
                reply.ansIntParam1=-1;
                reply.isValid=false;
                replyRequest(&reply);
            }
        }
    }
    out:
        printf("the sever is closed\n");
        interrupt->Halt();
}

